// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: Crop code for in-process communication
// ------------------------------------------------------------------

#include <edds/rtps/RTPSDomain.h>

#include <history/WriterHistory.h>
#include <edds/rtps/participant/RTPSParticipant.h>
#include <edds/rtps/reader/RTPSReader.h>
#include <edds/rtps/writer/RTPSWriter.h>

//#include <UDPv4Transport.h>
//#include <UDPv6Transport.h>

#include <deps/common/IPFinder.h>
#include <ertps/utils/LocatorAlias.h>
#include <deps/common/System.h>
#include <deps/common/md5.h>
#include <ertps/xmlparser/XMLProfileManager.h>
#include <rtps/participant/RTPSParticipantImpl.h>

#include <chrono>
#include <cstdlib>
#include <fstream>
#include <regex>
#include <string>
#include <thread>

#include <elog/Log.hpp>
#include <rtps/RTPSDomainImpl.hpp>

#include <rtps/common/GuidUtils.hpp>
#include <deps/common/Host.hpp>
#include <deps/common/SystemInfo.hpp>
#include <edds/rtps/attributes/PropertyPolicy.h>

#if HAVE_SECURITY
#include <security/OpenSSLInit.hpp>
#endif  // HAVE_SECURITY

namespace evbs {
namespace ertps {
namespace rtps {

static void guid_prefix_create(uint32_t ID, GuidPrefix_t& guidP) {
    evbs::edds::rtps::GuidUtils::instance().guid_prefix_create(ID, guidP);
}

#if HAVE_SECURITY
std::shared_ptr<security::OpenSSLInit> openssl_singleton_handler_ {security::OpenSSLInit::get_instance()};
#endif  // HAVE_SECURITY
std::mutex RTPSDomain::m_mutex;
std::atomic<uint32_t> RTPSDomain::m_maxRTPSParticipantID(1U);
std::vector<RTPSDomain::t_p_RTPSParticipant> RTPSDomain::m_RTPSParticipants;
std::set<uint32_t> RTPSDomain::m_RTPSParticipantIDs;
vbs::FileWatchHandle RTPSDomainImpl::file_watch_handle_;

void RTPSDomain::stopAll() {
    std::unique_lock<std::mutex> lock(m_mutex);
    logDebug(RTPS_PARTICIPANT, "DELETING ALL ENDPOINTS IN THIS DOMAIN");

    // Stop monitoring environment file
    vbs::SystemInfo::stop_watching_file(RTPSDomainImpl::file_watch_handle_);

    while (m_RTPSParticipants.size() > 0U) {
        RTPSDomain::t_p_RTPSParticipant participant = m_RTPSParticipants.back();
        (void)m_RTPSParticipantIDs.erase(m_RTPSParticipantIDs.find(participant.second->getRTPSParticipantID()));
        m_RTPSParticipants.pop_back();

        lock.unlock();
        RTPSDomain::removeRTPSParticipant_nts(participant);
        lock.lock();
    }
    logDebug(RTPS_PARTICIPANT, "RTPSParticipants deleted correctly ");
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
}

RTPSParticipant* RTPSDomain::createParticipant(uint32_t domain_id, const RTPSParticipantAttributes& attrs,
                                               vbs::common::ParticipantListener* listen) {
    return createParticipant(domain_id, true, attrs, listen);
}

RTPSParticipant* RTPSDomain::createParticipant(uint32_t domain_id, bool enabled, const RTPSParticipantAttributes& attrs,
                                               vbs::common::ParticipantListener* listen) {
    RTPSParticipantAttributes PParam = attrs;

    if ((PParam.builtin.discovery_config.leaseDuration < c_TimeInfinite) &&
        (PParam.builtin.discovery_config.leaseDuration <=
         PParam.builtin.discovery_config.leaseDuration_announcementperiod)) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "RTPSParticipant Attributes: LeaseDuration should be > "
                  "leaseDuration announcement period");
        return nullptr;
    }

    // Only the first time, initialize environment file watch if the corresponding environment
    // variable is set
    if ((!RTPSDomainImpl::file_watch_handle_)) {
        std::string filename = vbs::SystemInfo::get_environment_file();
        if ((!filename.empty() && (vbs::SystemInfo::file_exists(filename)))) {
            // Create filewatch
            RTPSDomainImpl::file_watch_handle_ =
                vbs::SystemInfo::watch_file(std::move(filename), RTPSDomainImpl::file_watch_callback);
        } else if ((!filename.empty())) {
            logWarning(RTPS_PARTICIPANT, filename + " does not exist. File watching not initialized.");
        }
    }

    uint32_t ID;
    {
        std::lock_guard<std::mutex> guard(m_mutex);

        if (PParam.participantID < 0) {
            ID = getNewId();
            while (m_RTPSParticipantIDs.insert(ID).second == false) {
                ID = getNewId();
            }
        } else {
            ID = static_cast<uint32_t>(PParam.participantID);
            if (m_RTPSParticipantIDs.insert(ID).second == false) {
                elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                          "RTPSParticipant with the same ID already exists, id " << PParam.participantID);
                return nullptr;
            }
        }
    }

    if ((!PParam.defaultUnicastLocatorList.isValid())) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Default Unicast Locator List contains invalid Locator");
        return nullptr;
    }
    if ((!PParam.defaultMulticastLocatorList.isValid())) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER,
                  "Default Multicast Locator List contains invalid Locator");
        return nullptr;
    }

    PParam.participantID = static_cast<int32_t>(ID);

    // Generate a new GuidPrefix_t
    GuidPrefix_t guidP;
    guid_prefix_create(ID, guidP);

    const std::string* guid_property = PropertyPolicyHelper::find_property(attrs.properties, "vbs.guidprefix");
    if (guid_property != nullptr) {
        uint32_t hash = evbs::ertps::rtps::e2eCommon::calculateCRC32P4(
            0, reinterpret_cast<const uint8_t*>(guid_property->data()), guid_property->size(), true);
        guidP.value[4] = static_cast<octet>(hash & 0xFF);
        guidP.value[5] = static_cast<octet>((hash >> 8) & 0xFF);
        guidP.value[6] = static_cast<octet>((hash >> 16) & 0xFF);
        guidP.value[7] = static_cast<octet>((hash >> 24) & 0xFF);
        guidP.value[11] |= 0x01U;
    }

    RTPSParticipant* p = new RTPSParticipant(nullptr);
    if (!p) {
        return nullptr;
    }
    RTPSParticipantImpl* pimpl = nullptr;

    // If we force the participant to have a specific prefix we must define a different persistence
    // GuidPrefix_t that would ensure builtin endpoints are able to differentiate between a
    // communication loss and a participant recovery
    if (PParam.prefix != c_GuidPrefix_Unknown) {
        pimpl = new RTPSParticipantImpl(domain_id, PParam, PParam.prefix, guidP, p, listen);
    } else {
        pimpl = new RTPSParticipantImpl(domain_id, PParam, guidP, p, listen);
    }

    // Check implementation was correctly initialized
    if ((!pimpl->is_initialized())) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR, "Cannot create participant due to initialization error");
        delete pimpl;
        return nullptr;
    }

    // Check there is at least one transport registered.
    if ((!pimpl->network_factory()->hasRegisteredTransports(pimpl->get_transport_name()))) {
        elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_PRECONDITION_NOT_MET,
                  "Cannot create participant, because there is any transport");
        delete pimpl;
        return nullptr;
    }

    {
        std::lock_guard<std::mutex> guard(m_mutex);
        m_RTPSParticipants.push_back(t_p_RTPSParticipant(p, pimpl));
    }

    // Check the environment file in case it was modified during participant creation leading to a
    // missed callback.
    if (RTPSDomainImpl::file_watch_handle_) {
        pimpl->environment_file_has_changed();
    }

    if (enabled) {
        // Start protocols
        pimpl->enable();
    }
    return p;
}

bool RTPSDomain::removeRTPSParticipant(RTPSParticipant* p) {
    if (p != nullptr) {
        std::unique_lock<std::mutex> lock(m_mutex);
        for (auto it = m_RTPSParticipants.begin(); it != m_RTPSParticipants.end(); ++it) {
            if (it->second->getGuid().guidPrefix == p->getGuid().guidPrefix) {
                RTPSDomain::t_p_RTPSParticipant participant = *it;
                (void)m_RTPSParticipants.erase(it);
                (void)m_RTPSParticipantIDs.erase(m_RTPSParticipantIDs.find(participant.second->getRTPSParticipantID()));
                lock.unlock();
                removeRTPSParticipant_nts(participant);
                return true;
            }
        }
    }
    elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_BAD_PARAMETER, "RTPSParticipant not valid or not recognized");
    return false;
}

void RTPSDomain::removeRTPSParticipant_nts(RTPSDomain::t_p_RTPSParticipant& participant) {
    participant.second->disable();
    // The destructor of RTPSParticipantImpl already deletes the associated RTPSParticipant and sets
    // its pointer to the RTPSParticipant to nullptr, so there is no need to do it here manually.
    delete (participant.second);
}

RTPSWriter* RTPSDomain::createRTPSWriter(RTPSParticipant* p, WriterAttributes& watt, WriterHistory* hist,
                                         WriterListener* listen) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSWriter* ret_val = nullptr;
        if (impl->createWriter(&ret_val, watt, hist, listen)) {
            return ret_val;
        }
    }

    return nullptr;
}

RTPSWriter* RTPSDomain::createRTPSWriter(RTPSParticipant* p, WriterAttributes& watt,
                                         const std::shared_ptr<IPayloadPool>& payload_pool, WriterHistory* hist,
                                         WriterListener* listen) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSWriter* ret_val = nullptr;
        if (impl->createWriter(&ret_val, watt, payload_pool, hist, listen)) {
            return ret_val;
        }
    }

    return nullptr;
}

RTPSWriter* RTPSDomain::createRTPSWriter(RTPSParticipant* p, WriterAttributes& watt,
                                         const std::shared_ptr<IPayloadPool>& payload_pool,
                                         const std::shared_ptr<IChangePool>& change_pool, WriterHistory* hist,
                                         WriterListener* listen) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSWriter* ret_val = nullptr;
        if (impl->createWriter(&ret_val, watt, payload_pool, change_pool, hist, listen)) {
            return ret_val;
        }
    }

    return nullptr;
}

RTPSWriter* RTPSDomain::createRTPSWriter(RTPSParticipant* p, const EntityId_t& entity_id, WriterAttributes& watt,
                                         const std::shared_ptr<IPayloadPool>& payload_pool,
                                         const std::shared_ptr<IChangePool>& change_pool, WriterHistory* hist,
                                         WriterListener* listen) {
    return RTPSDomainImpl::create_rtps_writer(p, entity_id, watt, payload_pool, change_pool, hist, listen);
}

RTPSWriter* RTPSDomain::createRTPSWriter(RTPSParticipant* p, const EntityId_t& entity_id, WriterAttributes& watt,
                                         const std::shared_ptr<IPayloadPool>& payload_pool, WriterHistory* hist,
                                         WriterListener* listen) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSWriter* ret_val = nullptr;
        if (impl->createWriter(&ret_val, watt, payload_pool, hist, listen, entity_id)) {
            return ret_val;
        }
    }

    return nullptr;
}

RTPSWriter* RTPSDomainImpl::create_rtps_writer(RTPSParticipant* p, const EntityId_t& entity_id, WriterAttributes& watt,
                                               const std::shared_ptr<IPayloadPool>& payload_pool,
                                               const std::shared_ptr<IChangePool>& change_pool, WriterHistory* hist,
                                               WriterListener* listen) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSWriter* ret_val = nullptr;
        if (impl->createWriter(&ret_val, watt, payload_pool, change_pool, hist, listen, entity_id)) {
            return ret_val;
        }
    }

    return nullptr;
}

bool RTPSDomain::removeRTPSWriter(RTPSWriter* writer) {
    if (writer != nullptr) {
        std::unique_lock<std::mutex> lock(m_mutex);
        for (auto it = m_RTPSParticipants.begin(); it != m_RTPSParticipants.end(); ++it) {
            if (it->first->getGuid().guidPrefix == writer->getGuid().guidPrefix) {
                t_p_RTPSParticipant participant = *it;
                lock.unlock();
                return participant.second->deleteUserEndpoint(writer->getGuid());
            }
        }
    }
    return false;
}

RTPSReader* RTPSDomain::createRTPSReader(RTPSParticipant* p, ReaderAttributes& ratt, ReaderHistory* rhist,
                                         ReaderListener* rlisten) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSReader* reader;
        if (impl->createReader(&reader, ratt, rhist, rlisten)) {
            return reader;
        }
    }
    return nullptr;
}

RTPSReader* RTPSDomain::createRTPSReader(RTPSParticipant* p, ReaderAttributes& ratt,
                                         const std::shared_ptr<IPayloadPool>& payload_pool, ReaderHistory* rhist,
                                         ReaderListener* rlisten) {
    RTPSParticipantImpl* impl = RTPSDomainImpl::find_local_participant(p->getGuid());
    if (impl) {
        RTPSReader* reader;
        if (impl->createReader(&reader, ratt, payload_pool, rhist, rlisten)) {
            return reader;
        }
    }
    return nullptr;
}

RTPSReader* RTPSDomain::createRTPSReader(RTPSParticipant* p, const EntityId_t& entity_id, ReaderAttributes& ratt,
                                         const std::shared_ptr<IChangePool>& change_pool, ReaderHistory* rhist,
                                         ReaderListener* rlisten) {
    RTPSParticipantImpl* impl = p->mp_impl;
    if (impl) {
        RTPSReader* reader;

        if (impl->createReader(&reader, ratt, change_pool, rhist, rlisten, entity_id)) {
            return reader;
        }
    }
    return nullptr;
}

bool RTPSDomain::removeRTPSReader(RTPSReader* reader) {
    if (reader != nullptr) {
        std::unique_lock<std::mutex> lock(m_mutex);
        for (auto it = m_RTPSParticipants.begin(); it != m_RTPSParticipants.end(); ++it) {
            if (it->first->getGuid().guidPrefix == reader->getGuid().guidPrefix) {
                t_p_RTPSParticipant participant = *it;
                lock.unlock();
                return participant.second->deleteUserEndpoint(reader->getGuid());
            }
        }
    }
    return false;
}

void RTPSDomainImpl::create_participant_guid(int32_t& participant_id, GUID_t& guid) {
    if (participant_id < 0) {
        std::lock_guard<std::mutex> guard(RTPSDomain::m_mutex);
        do {
            participant_id = static_cast<int32_t>(RTPSDomain::getNewId());
        } while (RTPSDomain::m_RTPSParticipantIDs.find(static_cast<uint32_t>(participant_id)) !=
                 RTPSDomain::m_RTPSParticipantIDs.end());
    }

    guid_prefix_create(participant_id, guid.guidPrefix);
    guid.entityId = c_EntityId_RTPSParticipant;
}

RTPSParticipantImpl* RTPSDomainImpl::find_local_participant(const GUID_t& guid) {
    std::lock_guard<std::mutex> guard(RTPSDomain::m_mutex);
    for (const RTPSDomain::t_p_RTPSParticipant& participant : RTPSDomain::m_RTPSParticipants) {
        if (participant.second->getGuid().guidPrefix == guid.guidPrefix) {
            // Participant found, forward the query
            return participant.second;
        }
    }

    return nullptr;
}

RTPSReader* RTPSDomainImpl::find_local_reader(const GUID_t& reader_guid) {
    std::lock_guard<std::mutex> guard(RTPSDomain::m_mutex);
    for (const RTPSDomain::t_p_RTPSParticipant& participant : RTPSDomain::m_RTPSParticipants) {
        if (participant.second->getGuid().guidPrefix == reader_guid.guidPrefix) {
            // Participant found, forward the query
            return participant.second->find_local_reader(reader_guid);
        }
    }

    return nullptr;
}

RTPSWriter* RTPSDomainImpl::find_local_writer(const GUID_t& writer_guid) {
    std::lock_guard<std::mutex> guard(RTPSDomain::m_mutex);
    for (const RTPSDomain::t_p_RTPSParticipant& participant : RTPSDomain::m_RTPSParticipants) {
        if (participant.second->getGuid().guidPrefix == writer_guid.guidPrefix) {
            // Participant found, forward the query
            return participant.second->find_local_writer(writer_guid);
        }
    }

    return nullptr;
}
#ifdef INTRA_PROCESS_ENABLE
/**
 * Check whether intraprocess delivery should be used between two GUIDs.
 *
 * @param local_guid    GUID of the local endpoint performing the query.
 * @param matched_guid  GUID being queried about.
 *
 * @returns true when intraprocess delivery is enabled, false otherwise.
 */
bool RTPSDomainImpl::should_intraprocess_between(const GUID_t& local_guid, const GUID_t& matched_guid) {
    uint32_t local_pid =
        local_guid.guidPrefix.is_guid_static() ? vbs::SystemInfo::instance().process_id() : local_guid.get_pid();
    if (!local_guid.is_on_same_process_as(local_pid, matched_guid, matched_guid.get_pid())) {
        // Not on the same process, should not use intraprocess mechanism.
        return false;
    }

    if (local_guid.entityId == c_EntityId_SPDPWriter || local_guid.entityId == c_EntityId_SPDPReader) {
        // Always disabled for PDP, to avoid inter-domain communications.
        return false;
    }

    switch (xmlparser::XMLProfileManager::library_settings().intraprocess_delivery) {
        case IntraprocessDeliveryType::INTRAPROCESS_FULL:
            return true;

        case IntraprocessDeliveryType::INTRAPROCESS_USER_DATA_ONLY:
            return !matched_guid.is_builtin();

        case IntraprocessDeliveryType::INTRAPROCESS_OFF:
        default:
            break;
    }

    return false;
}
#endif

void RTPSDomainImpl::file_watch_callback() {
    auto _1s = std::chrono::seconds(1);

    // Ensure that all changes have been saved by the OS
    (void)vbs::SystemInfo::wait_for_file_closure(vbs::SystemInfo::get_environment_file(), _1s);

    // For all RTPSParticipantImpl registered in the RTPSDomain, call
    // RTPSParticipantImpl::environment_file_has_changed
    std::lock_guard<std::mutex> guard(RTPSDomain::m_mutex);
    for (auto& participant : RTPSDomain::m_RTPSParticipants) {
        participant.second->environment_file_has_changed();
    }
}

}  // namespace rtps
}  // namespace ertps
}  // namespace evbs
